package com.atguigu.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.OrderDetail;
import com.atguigu.realtime.bean.OrderInfo;
import com.atguigu.realtime.bean.OrderWide;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.common.ConstantTopic;
import com.atguigu.realtime.util.DimQuery;
import com.atguigu.realtime.util.MyJDBCUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Map;

/**
 * @ClassName: DWMOrderWideApp
 * @Description:  V1.0     每次读取都是使用phoenix连接，查询hbase，效率低
 * @Author: kele
 * @Date: 2021/4/20 14:53
 *
 * 统计订单详情的宽表
 *
 * 1、两个事实表进行interval join     （涉及到两个流进行join，所以baseapp需要读取两个流，）
 * 2、将合并的事实表和维度表进行join，  （通过直接查询phoenix的方法读取）
 *
 *                      缺点：1、需要不断通过phoenix查询数据，来一条需要连接6次，
 *
 **/
public class DWMOrderWideApp extends BaseAppV2 {

    public static void main(String[] args) throws SQLException, ClassNotFoundException {

        new DWMOrderWideApp().init(30003,
                2,
                "DWMOrderWideApp",
                "DWMOrderWideApp",
                ConstantTopic.DWD_ORDER_INFO,
                "dwd_order_detail");
    }

    @Override
    protected void run(StreamExecutionEnvironment env, Map<String, DataStream<String>> streams) throws SQLException, ClassNotFoundException {

        //1、将订单表和订单详情表进行interval join（两个事实表）
        SingleOutputStreamOperator<OrderWide> orderWideStream = StreamJoin(streams);

        //2、与维度表进行join，通过查询phoenix进行join
        SingleOutputStreamOperator<OrderWide> orderWideWithDimStream = JoinDim(orderWideStream);

        orderWideWithDimStream.print();
    }

    /**
     * 与维度表进行关联
     * @param orderWideStream
     */
    private SingleOutputStreamOperator<OrderWide> JoinDim(SingleOutputStreamOperator<OrderWide> orderWideStream) throws SQLException, ClassNotFoundException {

        SingleOutputStreamOperator<OrderWide> result = orderWideStream

                .map(new RichMapFunction<OrderWide, OrderWide>() {

                    private Connection connect;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        connect = MyJDBCUtil.getConnect(Constant.PHOENIX_URL, Constant.PHOENIX_DRIVER);
                    }

                    @Override
                    public OrderWide map(OrderWide orderWide) throws Exception {

      /*          //维度表中只是通过id查询，可以将sql语句再次封装，可以不使用
                String sql = "select * from DIM_USER_INFO where id=?";
                Object[] args = {orderWide.getUser_id()};
                List<JSONObject> query = MyJDBCUtil.getQuery(connect,
                        sql,
                        args,
                        JSONObject.class
                );*/

                        //关联user_info表
                        JSONObject userInfo = DimQuery
                                .getDimQuery(connect, "DIM_USER_INFO", orderWide.getUser_id());
                        //在phoenix中，查询出的字段名是大写的
                        orderWide.setUser_gender(userInfo.getString("GENDER"));
                        orderWide.calcAge(userInfo.getString("BIRTHDAY"));


                        //关联dim_base_province表
                        JSONObject baseProvince = DimQuery
                                .getDimQuery(connect, "DIM_BASE_PROVINCE", orderWide.getProvince_id());
                        orderWide.setProvince_3166_2_code(baseProvince.getString("ISO_3166_2"));
                        orderWide.setProvince_area_code(baseProvince.getString("AREA_CODE"));
                        orderWide.setProvince_name(baseProvince.getString("NAME"));
                        orderWide.setProvince_iso_code(baseProvince.getString("ISO_CODE"));

                        //关联sku_info表
                        JSONObject skuInfo = DimQuery
                                .getDimQuery(connect, "DIM_SKU_INFO", orderWide.getSku_id());
                        orderWide.setSku_name(skuInfo.getString("SKU_NAME"));
                        orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
                        orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
                        orderWide.setTm_id(skuInfo.getLong("TM_ID"));

                        //关联spu_info表
                        JSONObject spuInfo = DimQuery
                                .getDimQuery(connect, "DIM_SPU_INFO", orderWide.getSpu_id());
                        orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));

                        //关联dim_base_category3表
                        JSONObject category3 = DimQuery
                                .getDimQuery(connect, "DIM_BASE_CATEGORY3", orderWide.getCategory3_id());
                        orderWide.setCategory3_name(category3.getString("NAME"));

                        //关联DIM_BASE_TRADEMARK表
                        JSONObject baseTrademark = DimQuery
                                .getDimQuery(connect, "DIM_BASE_TRADEMARK", orderWide.getTm_id());
                        orderWide.setTm_name(baseTrademark.getString("TM_NAME"));

                        return orderWide;
                    }

                    @Override
                    public void close() throws Exception {
                        connect.close();
                    }
                });

        return result;
    }

    /**
     * 将订单表和订单详情表进行join，得到宽表
     *
     * 使用interval join的方式进行join，
     *      --1、需要使用事件时间
     *      --2、必须是keyby之后的数据
     */
    private SingleOutputStreamOperator<OrderWide> StreamJoin(Map<String, DataStream<String>> streams) {

        //获取数据，封装为对象，添加水印，按照订单id进行keyby

        KeyedStream<OrderInfo, Long> orderInfoStream = streams
                .get("dwd_order_info")
                .map(s -> JSON.parseObject(s, OrderInfo.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((ele, ts) -> ele.getCreate_ts())
                )
                .keyBy(OrderInfo::getId);


        KeyedStream<OrderDetail, Long> orderDetailStream = streams
                .get("dwd_order_detail")
                .map(s -> JSON.parseObject(s, com.atguigu.realtime.bean.OrderDetail.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((ele, ts) -> ele.getCreate_ts())
                )
                .keyBy(OrderDetail::getOrder_id);

        SingleOutputStreamOperator<OrderWide> result = orderInfoStream
                .intervalJoin(orderDetailStream)
                .between(Time.seconds(-5), Time.seconds(5))
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                    @Override
                    public void processElement(OrderInfo orderInfo,
                                               OrderDetail orderDetail,
                                               Context ctx,
                                               Collector<OrderWide> out) throws Exception {

                        //将两个流join的结果作为OrderWide输出
                        out.collect(new OrderWide(orderInfo, orderDetail));
                    }
                });

        return result;
    }
}
